package com.xiaobai.spike.web;

import com.rabbitmq.client.Channel;
import com.xiaobai.spike.mapper.OrdersMapper;
import com.xiaobai.spike.mapper.ProductMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@RestController
@RequestMapping("/product")
public class ProductController {

    private static final Logger LOGGER = LoggerFactory.getLogger(ProductController.class);

    @Resource
    private OrdersMapper ordersMapper;

    @Resource
    private ProductMapper productMapper;

    @Autowired
    private RedisTemplate<String,Object> redisTemplate;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public Integer robbingProduct_mq(Integer id, String customer) {
        Map<String, Object> map = new HashMap<>(2);
        map.put("id",id);
        map.put("customer",customer);
        rabbitTemplate.convertAndSend("amq.direct","k1", map);
        return 0;
    }

    /**
     * @RabbitListener 可以标注在类上面，需配合 @RabbitHandler 注解一起使用
     * @RabbitListener 标注在类上面表示当有收到消息的时候，就交给 @RabbitHandler 的方法处理，具体使用哪个方法处理，
     * 根据 MessageConverter 转换后的参数类型
     *
     * 使用 @Payload 和 @Headers 注解可以消息中的 body 与 headers 信息
     *
     * 通过 ACK 确认是否被正确接收，每个 Message 都要被确认（acknowledged），可以手动去 ACK 或自动 ACK
     */
//    @RabbitListener(queues = "secondSkill") // 指定监听的队列名
    public void receiver(@Payload Map<String, Object> map, @Headers Channel channel, Message message) throws IOException {
        try {
            //处理消息
            if(this.robbingProduct_mysql((int)map.get("id"),String.valueOf(map.get("customer"))) == 1) {
                LOGGER.info("商品更新成功:{} - ", map.get("customer"));
            }
//             确认消息已经消费成功
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

        } catch (Exception e) {
            LOGGER.error("消费处理异常:{} - {}", map.get("customer"), e);
//             拒绝当前消息，并把消息返回原队列
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }
    public Integer robbingProduct_mysql(Integer id, String customer) {
        Integer lowCount = productMapper.getLowCount(id);
        if (0 < lowCount) {
            Integer insert = productMapper.success(id);
            if (insert == 1) {
                return ordersMapper.insert(id, customer);
            }
        }
        return 0;
    }

    @GetMapping("/spikes")
    public String spikes(@RequestParam Integer id, @RequestParam String customer) {
        String luaScript = "local userId = KEYS[1];\n" +
                "local stockKey = KEYS[2];\n" +
                "local userIdKey = KEYS[3];\n" +
                "local userExists = redis.call(\"sismember\", userIdKey, userId); \n" +
                "if tonumber(userExists)==1\n" +
                "then\n" +
                "  return 2;\n" +
                "end\n" +
                "local num = redis.call(\"get\" ,stockKey);\n" +
                "if tonumber(num) <= 0 then  return 0;\n" +
                "else\n" +
                " redis.call(\"decr\", stockKey);\n" +
                " redis.call(\"sadd\", userIdKey, userId);\n" +
                "end\n" +
                "return 1;";
        // 指定 lua 脚本，并且指定返回值类型
        // (为什么返回值不用 Integer 接收而是用 Long。这里是因为 spring-boot-starter-data-redis 提供的返回类型里面不支持 Integer。)
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>(luaScript, Long.class);
        List<String> keys = new ArrayList<>();
        keys.add(customer);
        keys.add(id + ":stock");
        keys.add(id + ":userId");
        // 参数一：redisScript，参数二：key列表，参数三：arg（可多个）
        Long result = redisTemplate.execute(redisScript, keys);
        if (0 == result) {
            return "秒杀结束了。。。";
        } else if (1 == result) {
            robbingProduct_mq(id, customer);
            return "恭喜你！秒杀成功了！";

        } else if (2 == result) {
            return "你已经秒杀成功了，不能重复秒杀";
        }
        return "秒杀异常啦~";
    }
}
